package com.ksvip.next.components.ons;

import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Properties;

import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.time.DateFormatUtils;
import org.springframework.beans.factory.InitializingBean;

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

/**
 * ons-client v1.8.4.Final 版本资源隔离优化（实例化支持）
 *
 * @author 2021年8月13日 上午10:52:16
 */
@Getter
@Setter
@Slf4j
public abstract class AbstractOnsSender implements InitializingBean {

    /**
     * ONS topic
     */
    protected String   topic;

    /**
     * ONS CID
     */
    protected String   groupId;
    /**
     * AccessKey
     */
    protected String   accessKey;
    /**
     * SecretKey
     */
    protected String   secretKey;
    /**
     * ONS Addr
     */
    protected String   namesrvAddr;

    protected String   defaultTag = "*";

    protected Producer producer;
    /**
     * ons的 延迟时间
     */
    protected int      delaySeconds;

    /**
     * 创建bean时，启动监听
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        log.info("开始启动消息监听");
        final Properties properties = new Properties();
        properties.put(PropertyKeyConst.GROUP_ID, this.groupId);
        properties.put(PropertyKeyConst.AccessKey, this.accessKey);
        properties.put(PropertyKeyConst.SecretKey, this.secretKey);
        // 推荐配置。
        properties.put(PropertyKeyConst.NAMESRV_ADDR, this.namesrvAddr);
        // 兼容配置（不推荐继续使用，建议逐渐升级为推荐配置）。
        //  properties.put(PropertyKeyConst.ONSAddr, "xxxx");
        Producer producer = ONSFactory.createProducer(properties);
        producer.start();
        this.producer = producer;
        log.info("OnsSender启动。。。。producer={}", producer);

    }

    /***
     * 发送消息
     *
     * @param tag 标签
     * @param msg 发送的信息
     * @return 发送结果
     */
    public SendResult sendMsg(final String tag, final String msg) {
        try {
            /**
             * 日期格式化字符串:yyyy-MM-dd HH:mm:ss
             */
            final String DATE_FORMAT_YYYY_MM_DD_HH_MM_SS = "yyyy-MM-dd HH:mm:ss";
            log.info("消息标签={};,时间={}", tag, DateFormatUtils.format(new Date(), DATE_FORMAT_YYYY_MM_DD_HH_MM_SS));
            return this.sendMsg(tag, msg.getBytes(StandardCharsets.UTF_8.name()));
        } catch (final UnsupportedEncodingException e) {
            log.error("发送消息失败", e);
            return null;
        }
    }

    /**
     * sendMsg
     *
     * @param tag
     * @param body
     * @return
     */
    public SendResult sendMsg(final String tag, final byte[] body) {
        if (StringUtils.isBlank(this.groupId) || StringUtils.isBlank(this.topic)
                || (StringUtils.isBlank(tag) && StringUtils.isBlank(this.defaultTag)) || ArrayUtils.isEmpty(body)) {
            throw new IllegalArgumentException("需要groupId、topic、tag、msg信息");
        }
        if (this.producer == null) {
            throw new IllegalArgumentException(String.format("[%s]非法，请检查groupId", this.groupId));
        }
        try {
            final Message sendMsg = new Message(this.topic, StringUtils.isNotBlank(tag) ? tag : this.defaultTag, body);
            return this.producer.send(sendMsg);
        } catch (final Exception e) {
            log.error("发送消息失败,消息：{},错误：{}", new String(body), e);
            throw new IllegalAccessError("发送消息失败,原因：" + e);
        }
    }

    /**
     * 延时发送消息
     *
     * @param tag
     * @param body
     * @return
     */
    public SendResult delayTimeSendMsg(final String tag, String body) {
        if (StringUtils.isBlank(this.groupId) || StringUtils.isBlank(this.topic)
                || (StringUtils.isBlank(tag) && StringUtils.isBlank(this.defaultTag)) || StringUtils.isBlank(body)) {
            throw new IllegalArgumentException("需要groupId、topic、tag、msg信息");
        }
        if (this.producer == null) {
            throw new IllegalArgumentException(String.format("[%s]非法，请检查groupId", this.groupId));
        }
        try {
            final Message sendMsg = new Message(this.topic, StringUtils.isNotBlank(tag) ? tag : this.defaultTag,
                    body.getBytes(StandardCharsets.UTF_8.name()));
            long delayTime = System.currentTimeMillis() + this.delaySeconds * 1000;
            sendMsg.setStartDeliverTime(delayTime);

            SendResult sendResult = this.producer.send(sendMsg);
            log.info("发送任务状态检查消息，messageId:{} " + sendResult.getMessageId());
            return sendResult;
        } catch (final Exception e) {
            log.error("发送消息失败,消息：{},错误：{}", new String(body), e);
            throw new IllegalAccessError("发送消息失败,原因：" + e);
        }
    }
}
